// Copyright 2012 Anno Langen. All Rights Reserved.

package org.anno;

import com.google.inject.Guice;
import com.google.inject.Inject;
import com.google.inject.Injector;

import org.apache.mahout.math.DenseVector;
import org.apache.mahout.math.function.PlusMult;

import java.util.Arrays;

/**
 * Computes a low rank approximation of the differences from movie mean for the score matrix.
 *
 * @author arl@google.com (Anno Langen)
 */
public class PrincipalComponents {

  private final Ratings ratings;
  private final Movies movies;

  @Inject
  public PrincipalComponents(Ratings ratings, Movies movies) {
    this.ratings = ratings;
    this.movies = movies;
  }

  double[][] computeMovieComponents(int k) {
    double[][] result = new double[k][Movies.MAX_ID + 1];
    for (int i = 0; i < k; i++) {
      computeRankOne(i, result);
    }
    return result;
  }

  private void computeRankOne(int k, double[][] allM) {
    double[] m = allM[k];
    for (int i = 0; i < m.length; i++) {
      m[i] = 1;
    }
    orthogonalize(m, k, allM);

    for (int i = 20; --i >= 0; ) {
      zigzag(m);
      orthogonalize(m, k, allM);
    }
  }

  /**
   * Compute optimal user components for given movie components and compute optimal movie components
   * for the computed user components.
   */
  private void zigzag(final double[] m) {
    try {
      final char[][] shardUserIndex = new char[41][1 << 16];
      final double[][] userComponent = new double[41][Users.SHARD_USER_MAX];

      ratings.sendByUser(new Ratings.ProcessorFactory<Ratings.UserProcessor>() {

        class Shard {

          int userCount;
          final double[] shardDenominator = new double[Users.SHARD_USER_MAX];
          final double[] shardNominator;
          final char[] userIndex;

          Shard(int shardId) {
            shardNominator = userComponent[shardId];
            userIndex = shardUserIndex[shardId];
          }

          Ratings.UserProcessor get(int userId) {
            final int index = userCount++;
            userIndex[userId & 0xffff] = (char) index;

            return new Ratings.UserProcessor() {

              public void close() {
                shardNominator[index] /= shardDenominator[index];
              }

              public void movieScore(int movie, int score) {
                double v = m[movie];
                shardNominator[index] +=
                    (score - PrincipalComponents.this.movies.getMeanScore(movie)) * v;
                shardDenominator[index] += v * v;
              }
            };
          }
        }

        final Shard[] shards = new Shard[41];

        public Ratings.UserProcessor get(int id) {
          return getShard(id).get(id);
        }

        private Shard getShard(int id) {
          int shardId = id >> 16;
          Shard shard = shards[shardId];
          return shard != null ? shard : (shards[shardId] = new Shard(shardId));
        }
      });

      Arrays.fill(m, 0); // m acts as nominator    
      ratings.sendByMovie(new Ratings.ProcessorFactory<Ratings.MovieProcessor>() {
        final double[] denominator = new double[m.length];

        public Ratings.MovieProcessor get(final int movieId) {
          return new Ratings.MovieProcessor() {
            public void close() {
              m[movieId] /= denominator[movieId];
            }

            public void userScore(int user, int score) {
              int userShard = user >> 16;
              double u = userComponent[userShard][shardUserIndex[userShard][user & 0xffff]];
              m[movieId] += (score - movies.getMeanScore(movieId)) * u;
              denominator[movieId] += u * u;
            }
          };
        }
      });
    } catch (InterruptedException e) {
      throw new AssertionError(e);
    }
  }

  /**
   * Make vector m orthogonal to allM[0]..allM[k-1] by subtracting projection (Gramm Schmidt).
   */
  private void orthogonalize(double[] m, int k, double[][] allM) {
    DenseVector mV = new DenseVector(m, true);
    for (int i = 0; i < k; i++) {
      DenseVector iV = new DenseVector(allM[i], true);
      mV.assign(iV, new PlusMult(-mV.dot(iV) / Math.sqrt(iV.dotSelf()) / Math.sqrt(mV.dotSelf())));
    }
  }

  public static void main(String[] args) {
    Injector injector = Guice.createInjector(Ratings.getModule(), Movies.getModule());
    double[][] pc = injector.getInstance(PrincipalComponents.class).computeMovieComponents(2);
    Movie[] values = injector.getInstance(Movies.class).values;
    System.out.println("Title,PC1,PC2");
    for (int i = values.length; --i>=0;) {
      Movie movie = values[i];
      System.out.println(
          "\"" + movie.title.replaceAll("\"", "\"\"") + "\"," + pc[0][i] + "," + pc[1][i]);
    }
    Parallel.EXECUTOR.shutdown();
  }
}
